package com.flink.sync.oracle;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;


public class OracleDataSource extends RichSourceFunction<TQrUser> {
	private PreparedStatement ps;
	private Connection connection;

	/**
	 * open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接。
	 * @param parameters
	 * @throws Exception
	 */
	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		connection = getConnection();
		String sql = "select * from system.t_qr_user_rs where id not in (select id from osoadata.t_qr_user);";
		ps = this.connection.prepareStatement(sql);
	}

	private Connection getConnection() {
		Connection conn = null;
		try {
			Class.forName("oracle.jdbc.OracleDriver");
			conn = DriverManager.getConnection("jdbc:oracle:thin:@10.1.171.84:1521:ORCL", "osoadata", "888888");
		} catch (Exception e) {
			System.out.println("-----------oracle get connection has exception , msg = " + e.getMessage());
		}
		return conn;

	}

	/**
	 * 程序执行完毕就可以进行，关闭连接和释放资源的动作了
	 * @throws Exception
	 */
	@Override
	public void close() throws Exception {
		super.close();
		if (null != connection) {
			connection.close();
		}
		if (null != ps) {
			ps.close();
		}

	}

	/**
	 * DataStream 调用一次 run() 方法用来获取数据
	 * @param sourceContext
	 * @throws Exception
	 */
	@Override
	public void run(SourceContext<TQrUser> sourceContext) throws Exception {
		ResultSet resultSet = ps.executeQuery();
		while (resultSet.next()) {
			TQrUser tQrUser = new TQrUser();
			tQrUser.setId(resultSet.getLong("id"));
			tQrUser.setIdno(resultSet.getString("idno"));
			tQrUser.setMobileno(resultSet.getString("mobileno"));
			tQrUser.setQrcode(resultSet.getString("qrcode"));
			tQrUser.setStatus(resultSet.getString("status"));
			tQrUser.setModify_date(resultSet.getString("modify_date"));
			tQrUser.setName(resultSet.getString("name"));
			tQrUser.setYxwd(resultSet.getString("yxwd"));
			tQrUser.setCreate_date(resultSet.getString("create_date"));
			sourceContext.collect(tQrUser);
		}
	}

	@Override
	public void cancel() {

	}
}
